use std::future::poll_fn;
use std::pin::Pin;

use crate::openssl::stream::SslStream;
use crate::runtime::{RuntimeManager, SHUTDOWN_TOKEN};
use async_trait::async_trait;
use base::conn::{BoxHttpConn, HttpConn};
use log::{debug, warn};
use openssl::ssl::Ssl;
use proxy::HttpProxyWrap;
use tokio::select;

use crate::config::http::main::get_http_main_runtime_conf;
use crate::error::*;
use crate::{iface::AppHandle, listener::ArcListener, stream::Stream};

pub mod base;
pub mod pool;
pub mod proxy;
pub mod v1;
pub mod v2;

pub struct HttpServerApp {}

#[async_trait]
impl AppHandle for HttpServerApp {
    async fn handle(&self, stream: Stream, listener: ArcListener) {
        let hc = match self.do_handshake(stream, listener).await {
            Ok(raw_hc) => Box::new(raw_hc),
            Err(e) => {
                warn!("{}", e);
                return;
            }
        };

        // TODO: 处理http2接入.
        // TODO: SSL ALPN协商.
        let switch_h2 = false;
        if switch_h2 {
            self.h2_process(hc).await;
            return;
        }

        let proxy_wrap = HttpProxyWrap::new(hc);
        let mut reuse_hc = proxy_wrap.process_new().await;
        while let Some(hc) = reuse_hc {
            let proxy_wrap = HttpProxyWrap::new(hc);
            reuse_hc = proxy_wrap.process_new().await;
        }
    }
}

impl HttpServerApp {
    async fn do_handshake(
        &self,
        stream: Stream,
        listener: ArcListener,
    ) -> Result<HttpConn> {
        let listener_conf = listener.get_conf();

        if listener_conf.tls_enable {
            let main_conf = get_http_main_runtime_conf();
            let acceptor_ctx = &main_conf.ssl_acceptor_ctx;
            let peer_addr_str = stream.peer_addr_str().to_string();
            let local_addr_str = stream.local_addr_str().to_string();

            let ssl = Ssl::new(acceptor_ctx).if_err(
                ErrorType::HandshakeError,
                format!(
                    "while in create ssl object, client: {}, server: {}",
                    peer_addr_str, local_addr_str
                ),
            )?;

            let mut ssl_stream = SslStream::new(ssl, stream).if_err(
                ErrorType::HandshakeError,
                format!(
                    "while in create ssl object, client: {}, server: {}",
                    peer_addr_str, local_addr_str
                ),
            )?;

            Pin::new(&mut ssl_stream).accept().await.if_err(
                ErrorType::HandshakeError,
                format!(
                    "while in ssl stream accelt, client: {}, server: {}",
                    peer_addr_str, local_addr_str
                ),
            )?;

            return Ok(HttpConn::new_with_ssl_stream(
                ssl_stream,
                Some(listener),
            ));
        }

        Ok(HttpConn::new_with_stream(stream, Some(listener)))
    }

    async fn h2_process(&self, hc: BoxHttpConn) {
        let listener_clone = hc.listener.clone();

        // TODO: 经过ssl协商后, 此处就已经知道server conf了.
        let h2_builder = &get_http_main_runtime_conf()
            .h2_server_runtime_conf
            .h2_server_builder;
        let mut h2c =
            match v2::server_session::handshake(hc, Some(h2_builder)).await {
                Ok(h2c) => h2c,
                Err(e) => {
                    warn!("{}", e);
                    return;
                }
            };

        let shutdown_token = SHUTDOWN_TOKEN.clone();

        loop {
            let res = select! {
                _ = shutdown_token.cancelled() => {
                    h2c.graceful_shutdown();
                    if let Err(e) = poll_fn(|cx| h2c.poll_closed(cx)).await {
                        warn!("{} while h2c graceful_shutdown", e);
                    }
                    return;
                }

                res =  h2c.accept() => res,
            };

            let stream = match res {
                Some(res) => match res {
                    Ok(stream) => stream,
                    Err(e) => {
                        warn!("{} while in h2c accept stream", e);
                        return;
                    }
                },
                None => {
                    debug!("h2c closed");
                    return;
                }
            };

            let listener = listener_clone.clone();
            RuntimeManager::get_worker_runtime().get_handle().spawn(
                async move {
                    // TODO: 在h2c上统计request个数.
                    let requests = 0;
                    let hc = HttpConn::new_with_h2(
                        stream.0, stream.1, requests, listener,
                    );
                    let hc = Box::new(hc);

                    let proxy_wrap = HttpProxyWrap::new(hc);
                    proxy_wrap.process_new().await;
                },
            );
        }
    }
}
